#include "StdAfx.h"
#include "ParallelSystem.h"

#include <ppl.h>

_PARALLEL_SYSTEM_::CParallelSystem theParallelSystem;
_PARALLEL_SYSTEM_::CParallelSystem* CSingletonT<_PARALLEL_SYSTEM_::CParallelSystem>::s_pObject = nullptr;

namespace _PARALLEL_SYSTEM_
{
	///////////////////////////////////////////////////////////////////////////
	///////////////////////////////////////////////////////////////////////////

	void JobBase::PostJob( CParallelTask* task )
	{
		task->PostJob( this );
	}

	//dynamic job
	void DynamicJob::DoWork( CParallelWorker* worker )
	{
		Concurrency::task_group group;

		auto it = m_Jobs.begin();
		for( ; it != m_Jobs.end() ; ++it )
		{
			ParallelJobFunction& job = *it;
			group.run( job );
		}

		group.wait();
	}

	//ParallelJobEntry
	void ParallelJobEntry::AddJob( JobBase* job )
	{
		m_Jobs.push_back( job );
	}

	void ParallelJobEntry::AddJob( ParallelJobFunction& job )
	{
		Job* new_job = new Job;
		new_job->AddJob( job );

		m_Jobs.push_back( new_job );
	}

	void ParallelJobEntry::PostJob( CParallelTask* task )
	{
		for each( JobBase* job in m_Jobs )
			task->PostJob( job );
	}
	
	///////////////////////////////////////////////////////////////////////////
	///////////////////////////////////////////////////////////////////////////
	CParallelTask::CParallelTask()
	{
		m_task_id = 0;
	}

	CParallelTask::~CParallelTask()
	{
		delete m_queueJobs;
	}

	void CParallelTask::Initialize( BYTE task_id, size_t worker_count )
	{
		m_task_id = task_id;

		m_queueJobs = new CQueMT<JobBase*>(static_cast<long>(worker_count));

		m_workers.resize( worker_count );
		std::fill( m_workers.begin(), m_workers.end(), (CParallelWorker*)nullptr );

		for( unsigned long id = 1 ; id <= worker_count ; ++id )
		{
			CParallelWorker* worker = new CParallelWorker( this, id );
			worker->Create( worker, (1024 * 1024 * 2), THREAD_PRIORITY_NORMAL );
			worker->Start();

			m_workers[ id - 1 ] = worker;
		}
	}

	void CParallelTask::Finalize()
	{
		m_queueJobs->Deactivate();

		for each( CParallelWorker* worker in m_workers )
			delete worker;

		m_workers.clear();
	}

	void CParallelTask::PostJob( JobBase* job )
	{
		m_queueJobs->Enque(job);
	}

	void CParallelWorker::ThreadProc( LPVOID pParam )
	{
		JobBase* job = nullptr;
		while( true )
		{
			m_ownTask->m_queueJobs->Deque( &job );
			if( job == nullptr )
				break;

			job->DoWork( this );

			PARALLEL_SYSTEM.JobCompleted( job );
		}
	}

	unsigned long CParallelWorker::GetThreadID()
	{
		return m_worker_id -1;
	}

	///////////////////////////////////////////////////////////////////////////
	///////////////////////////////////////////////////////////////////////////
	CParallelSystem::CParallelSystem() : m_CompletedJobCS(TRUE)
	{
		static_task = nullptr;
	}

	CParallelSystem::~CParallelSystem()
	{

	}

	void CParallelSystem::Initialize()
	{
		SAFE_DELETE( static_task );

		static_task = new CParallelTask;

		SYSTEM_INFO sysinfo;
		::GetSystemInfo(&sysinfo);
		static_task->Initialize( STATIC_TASK_ID, sysinfo.dwNumberOfProcessors );
	}

	bool CParallelSystem::AddTask( CParallelTask* task )
	{
		BYTE task_id = task->GetTaskID();

		if( task_id == STATIC_TASK_ID || m_tasks[task_id] != nullptr )
			return false;

		m_tasks[ task_id ] = task;
		return true;
	}

	bool CParallelSystem::CreateDBTask( BYTE task_id, size_t session_count )
	{
		if( m_tasks.find(task_id) != m_tasks.end() )
			return false;

		CParallelTask* task = new CParallelTask;
		task->Initialize( task_id, session_count );

		return AddTask( task );
	}

	void CParallelSystem::Finalize()
	{
		for each( auto _value in m_tasks )
		{
			_value.second->Finalize();
			delete _value.second;
		}

		m_tasks.clear();

		if( static_task != nullptr )
		{
			static_task->Finalize();
			SAFE_DELETE( static_task );
		}
	}

	void CParallelSystem::PostJob( JobBase* job )
	{
		if( static_task != nullptr )
			static_task->PostJob( job );
	}

	void CParallelSystem::PostJob( BYTE task_id, JobBase* job )
	{
		auto it = m_tasks.find( task_id );
		if( it == m_tasks.end() )
			return;

		CParallelTask* task = it->second;
		job->PostJob( task );
	}

	void CParallelSystem::JobCompleted( JobBase* job )
	{
		m_CompletedJobCS.Lock();
		m_CompletedJobs.push( job );
		m_CompletedJobCS.Unlock();
	}

	void CParallelSystem::ProcessJobCompletion()
	{
		if( m_CompletedJobs.empty() == true )
			return;
		
		QUEUE_COMPLETED_JOBS completed_jobs;

		m_CompletedJobCS.Lock();
		m_CompletedJobs.swap( completed_jobs );
		m_CompletedJobCS.Unlock();

		while( completed_jobs.empty() == false )
		{
			JobBase* job = completed_jobs.front();

			JobBase::CompletionFunction& completion = job->m_completionHandler;
			if( completion != nullptr )
				completion( job );

			completed_jobs.pop();
			delete job;
		}
	}
};
